/*
 * ============================================================================
 *
 *       Filename:  IPRecordThreadReader.cpp
 *
 *    Description:  multi-thread reading and writing in windows, but it's more slowly
 *                  than single thread in my testing
 *
 *         Author:  liangqing
 *
 * ============================================================================
 */
#include "IPRecordThreadReader.h"

#ifdef WIN32
IPRecordThreadReader::IPRecordThreadReader(const char *filename, int size):
                                            IPRecordReaderQueue(filename),
                                            _size(size) {
    _pool = new IPRecord[_size];
    _write = -1;
    _read = -1;
    _finish = false;
    _resource_sem = CreateSemaphore(NULL, 0, _size, NULL);
    _space_sem = CreateSemaphore(NULL, _size, _size, NULL);
    _push_sem = CreateSemaphore(NULL, 0, _size, NULL);
    InitializeCriticalSection(&_critic);
    _thread = (HANDLE)_beginthreadex(NULL, 0, (IPRecordThreadReader::_run), this, 0, &_thread_id);
}

IPRecordThreadReader::~IPRecordThreadReader() {
    if(_pool)
        delete []_pool;
}

unsigned __stdcall IPRecordThreadReader::_run(void *arg) {
    IPRecordThreadReader *pReader = (IPRecordThreadReader *)arg;
    IPRecord __record;
    
    WaitForSingleObject(pReader->_push_sem, INFINITE);
    while(pReader->_get(&__record)) {
        WaitForSingleObject(pReader->_space_sem, INFINITE);
        EnterCriticalSection(&pReader->_critic);
        pReader->_write = (pReader->_write+1) % pReader->_size;
        pReader->_pool[pReader->_write] = __record;
        LeaveCriticalSection(&pReader->_critic);
        ReleaseSemaphore(pReader->_resource_sem, 1, NULL);
    }

    EnterCriticalSection(&pReader->_critic);
    pReader->_finish = true;
    LeaveCriticalSection(&pReader->_critic);
	_endthreadex(0);
    return 0;
}

bool IPRecordThreadReader::get(IPRecord *record) {
    bool s;
    EnterCriticalSection(&_critic);
    s = _finish;
    LeaveCriticalSection(&_critic);
    
    if(!s)
        WaitForSingleObject(_resource_sem, INFINITE);

    EnterCriticalSection(&_critic);
    
    _read = (_read+1) % _size;
    if(_finish && (_read == _write+1)) {
        LeaveCriticalSection(&_critic);
        return false;
    }
    (*record) = _pool[_read];

    LeaveCriticalSection(&_critic);
    if(!s)
        ReleaseSemaphore(_space_sem, 1, NULL);

    return true;
}

void IPRecordThreadReader::push_reader(IPRecordReaderType type, const char *filename) {
    _push_reader(type, filename);
    ReleaseSemaphore(_push_sem, 1, NULL);
}
#endif /* WIN32 */
